import net, { Socket, NetConnectOpts } from "net";
import { Pipe } from "./Pipe";
import { Logger } from "./Logger";
import { injectSymbol } from "./InjectSocket";

let rpcRequestId = 1;

const TIME_OUT = 5000;

export interface IRpcBase {
  meta: {
    type: RpcTypeEnum;
  };
}

export interface IRpcReq extends IRpcBase {
  name: string;
  params: any[];
  meta: {
    id: number;
    type: RpcTypeEnum.Req;
    resolve: (value: Pick<IRpcRes, "data" | "error">) => void;
  };
}

export interface IRpcRes extends IRpcBase {
  data: any;
  error?: string;
  meta: {
    id: number;
    type: RpcTypeEnum.Res;
  };
}

export interface IRpcMsg extends IRpcBase {
  name: string;
  params: any[];
  meta: {
    type: RpcTypeEnum.Msg;
  };
}

enum RpcTypeEnum {
  Req,
  Res,
  Msg,
}

interface IRpcConnectionOptions {
  socket?: Socket;
  netOptions?: NetConnectOpts;
  implement: any;
  logger: Logger;
}

interface IRpcServerOptions {
  port: number;
  implement: any;
  logger: Logger;
}

/***
 * Rpc服务端
 */
export class RpcServer {
  static async Create(options: IRpcServerOptions) {
    const server = new RpcServer(options);
    await server.init();
    return server;
  }

  constructor(private options: IRpcServerOptions) {}

  private init() {
    return new Promise((resolve) => {
      const server = net.createServer();

      server.listen(this.options.port, "localhost", () => {
        resolve(true);
      });

      server.on("error", (e) => {
        this.options.logger.error(e);
      });

      server.on("connection", async (socket) => {
        await RpcConnection.Create({
          socket,
          implement: this.options.implement,
          logger: this.options.logger,
        });
      });
    });
  }
}

/***
 * 通过RpcConnection封装主动和被动连接产生的socket对象
 */
export class RpcConnection {
  socket: Socket;
  logger: Logger;
  implement: any;
  rpcInvokingMap: Map<number, IRpcReq> = new Map();
  pipe = new Pipe();

  static async Create(options: IRpcConnectionOptions) {
    const client = new RpcConnection(options);
    await client.init();
    return client;
  }

  private constructor(private options: IRpcConnectionOptions) {
    this.logger = options.logger;
    this.implement = options.implement;
  }

  /***
   * 确定socket对象
   */
  private init() {
    return new Promise<void>((resolve, reject) => {
      if (this.options.socket) {
        this.socket = this.options.socket;
        this.bind();
        resolve();
      } else if (this.options.netOptions) {
        this.socket = net.createConnection(this.options.netOptions, () => {
          this.bind();
          resolve();
        });
      } else {
        this.logger.fatal("初始化异常");
        reject("初始化异常");
      }
    });
  }

  /***
   * 监听事件
   */
  private bind() {
    this.pipe.on("data", (buffer: Buffer) => {
      const data = JSON.parse(buffer.toString()) as IRpcBase;
      const { type } = data.meta;
      switch (type) {
        case RpcTypeEnum.Req:
          this.handleRpcReq(data as IRpcReq);
          break;
        case RpcTypeEnum.Res:
          this.handleRpcRes(data as IRpcRes);
          break;
        case RpcTypeEnum.Msg:
          this.handleRpcMsg(data as IRpcMsg);
          break;
        default:
          const _: never = type;
          break;
      }
    });

    this.socket.on("data", (buffer) => {
      this.pipe.put(buffer);
    });

    this.socket.on("close", () => {
      this.logger.fatal("Socket Close");
    });

    this.socket.on("error", (...args) => {
      this.logger.fatal("Socket", ...args);
    });
  }

  /***
   * RPC调用
   */
  call(params: Omit<IRpcReq, "meta">) {
    return new Promise<Pick<IRpcRes, "data" | "error">>((resolve) => {
      // 防止rpcRequestId过大
      if (rpcRequestId >= 2 ** 30) {
        rpcRequestId = 1;
      }
      const id = rpcRequestId++;
      const data: IRpcReq = {
        ...params,
        meta: {
          id,
          type: RpcTypeEnum.Req,
          resolve,
        },
      };
      const buffer = this.pipe.pack(data);
      this.socket.write(buffer);
      this.rpcInvokingMap.set(id, data);

      setTimeout(() => {
        resolve({
          error: "time out",
          data: null,
        });
        this.rpcInvokingMap.delete(id);
      }, TIME_OUT);
    });
  }

  /***
   * RPC消息
   */
  send(params: Omit<IRpcMsg, "meta">) {
    const data = {
      ...params,
      meta: {
        type: RpcTypeEnum.Msg,
      },
    };
    const buffer = this.pipe.pack(data);
    this.socket.write(buffer);
  }

  /***
   * Rpc被调用方处理调用指定函数并返回结果
   */
  private async handleRpcReq(req: IRpcReq) {
    const {
      name,
      params,
      meta: { id },
    } = req;

    try {
      const func = this.implement[name];
      if (func) {
        if (func[injectSymbol]) {
          params.unshift(this);
        }

        const data = await func.apply(this.implement, params);
        const res: IRpcRes = {
          data,
          meta: {
            id,
            type: RpcTypeEnum.Res,
          },
        };
        const buffer = this.pipe.pack(res);
        this.socket.write(buffer);
      } else {
        throw new Error(`func not exist: ${name}`);
      }
    } catch (e) {
      this.logger.error("handleRpcReq", e);
      const res: IRpcRes = {
        data: {},
        error: (e as Error).message,
        meta: {
          id,
          type: RpcTypeEnum.Res,
        },
      };
      const buffer = this.pipe.pack(res);
      this.socket.write(buffer);
    }
  }

  /***
   * Rpc调用方通过promise把结果resolve出去
   */
  private handleRpcRes(res: IRpcRes) {
    const {
      data,
      error,
      meta: { id },
    } = res;
    const invokingItem = this.rpcInvokingMap.get(id);
    if (invokingItem) {
      invokingItem.meta.resolve({
        data,
        error,
      });
      this.rpcInvokingMap.delete(id);
    } else {
      this.logger.warn("handleRpcRes invokingItem not exist:", JSON.stringify(res));
    }
  }

  private async handleRpcMsg(msg: IRpcMsg) {
    const { name, params } = msg;
    const func = this.implement[name];

    try {
      if (func) {
        await func.apply(this.implement, params);
      } else {
        throw new Error(`handleRpcMsg func not exist: ${name}`);
      }
    } catch (e) {
      this.logger.error(`handleRpcMsg error:`, e);
    }
  }
}
